package com.web.mqtt.service.impl;

import java.nio.charset.StandardCharsets;
import java.time.Instant;
import java.time.LocalDateTime;
import java.time.LocalTime;
import java.time.ZoneId;
import java.time.format.DateTimeFormatter;
import java.util.Base64;
import java.util.List;
import java.util.UUID;
import java.util.regex.Pattern;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.HttpMethod;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;
import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.DeserializationFeature;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
import com.fasterxml.jackson.datatype.jsr310.ser.LocalDateTimeSerializer;
import com.web.common.basic.api.ElasticsearchApi;
import com.web.common.basic.elasticsearch.entity.ElasticDataListParam;
import com.web.common.basic.elasticsearch.entity.ElasticParam;
import com.web.common.basic.entity.ElasticsearchIndex;
import com.web.common.basic.entity.SimplePage;
import com.web.common.iot.mqtt.entity.MqttBaseListener;
import com.web.common.iot.mqtt.entity.MqttBaseListenerEntity;
import com.web.common.iot.mqtt.entity.MqttConnectedListener;
import com.web.common.iot.mqtt.entity.MqttDeliveredListener;
import com.web.common.iot.mqtt.entity.MqttDisconnectedListener;
import com.web.common.iot.mqtt.entity.MqttDroppedListener;
import com.web.common.iot.mqtt.entity.MqttMessageType;
import com.web.common.iot.mqtt.entity.MqttSubscribedListener;
import com.web.common.iot.mqtt.entity.MqttUser;
import com.web.common.result.R;
import com.web.common.result.ResultEnum;
import com.web.common.security.FeignInnerEntity;
import com.web.common.tool.DateTimeUtil;
import com.web.common.tool.Symbols;
import com.web.mqtt.mapper.MqttUserMapper;
import com.web.service.handler.error.CustomException;

@Service
public class MqttListenServiceImpl {

	private static final ObjectMapper MAPPER = new ObjectMapper();
	
	private static final Pattern PATTERN = Pattern.compile("\r|\n");
	
	@Autowired
	private ElasticsearchApi elasticsearchApi;
	
	@Autowired
	private MqttUserMapper mqttUserMapper;
	
	static {
		JavaTimeModule module = new JavaTimeModule();
		module.addSerializer(LocalDateTime.class, new LocalDateTimeSerializer(DateTimeFormatter.ISO_DATE_TIME));
		MAPPER.registerModule(module);
		MAPPER.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
	}
	
	/**
	 * 获取MQTT日志列表
	 * @param condition 查询条件
	 * @return R<IPage<MqttBaseListener>> MQTT日志列表
	 *
	 * @author zhouhui
	 * @since 2024.04.27
	 */
	public R<IPage<MqttBaseListenerEntity>> getMqttLogList(MqttBaseListenerEntity condition) {
		ElasticDataListParam param = new ElasticDataListParam();
		param.setIndex(ElasticsearchIndex.MQTT_LOG_INDEX);
		param.setPage(condition.getPage());
		param.setSize(condition.getSize());
		param.setJsonData(setSearchJsonData(condition));
		R<SimplePage> result = elasticsearchApi.getDataList(param);
		if(!ResultEnum.SUCCESS.getCode().equals(result.getCode())) {
			return R.fail(null, result.getMessage());
		}
		
		SimplePage data = result.getData();
		List<MqttBaseListenerEntity> dataList = null;
		try {
			dataList = MAPPER.readValue(data.getRecords(), new TypeReference<List<MqttBaseListenerEntity>>() {});
			for(MqttBaseListenerEntity item: dataList) {
				if(StringUtils.hasText(item.getMessageType())) {
					item.setMessageTypeName(MqttMessageType.getMessageByType(item.getMessageType()));
				}
				item.setCreateTime(LocalDateTime.parse(item.getCreateTime()).format(DateTimeFormatter.ofPattern(DateTimeUtil.YEAR_MONTH_DATE_TIME)));
			}
		} catch (Exception e) {
			e.printStackTrace();
		}
		IPage<MqttBaseListenerEntity> pageResult = new Page<>(data.getPage(), data.getSize());
		pageResult.setTotal(data.getTotal());
		pageResult.setRecords(dataList);
		return R.ok(pageResult);
	}
	
	/**
	 * 设置查询的JSON数据
	 * @param condition 查询条件
	 * @return String JSON字符串
	 *
	 * @author zhouhui
	 * @since 2024.04.27
	 */
	private String setSearchJsonData(MqttBaseListenerEntity condition) {
		long start = condition.getStartDate().atTime(LocalTime.MIN).atZone(ZoneId.systemDefault()).toInstant().toEpochMilli();
		long end = condition.getEndDate().atTime(LocalTime.MAX).atZone(ZoneId.systemDefault()).toInstant().toEpochMilli();
		StringBuilder bd = new StringBuilder();
		bd.append("{");
		bd.append("\"query\": ");
		bd.append("{\"bool\": { \"must\":");
		bd.append("[");
		bd.append("{\"range\":{\"timestamp\":{\"gte\":"+ start +", \"lte\": "+ end +"}}}");
		if(StringUtils.hasText(condition.getClientId())) {
			bd.append(",{\"match\":{\"clientId\":\""+ condition.getClientId() +"\"}}");
		}
		if(StringUtils.hasText(condition.getUsername())) {
			bd.append(",{\"match\":{\"username\":\""+ condition.getUsername() +"\"}}");
		}
		if(StringUtils.hasText(condition.getMessageType())) {
			bd.append(",{\"match\":{\"messageType\":\""+ condition.getMessageType() +"\"}}");
		}
		if(StringUtils.hasText(condition.getTopic())) {
			bd.append(",{\"match\":{\"topic\":{\"query\":\""+ condition.getTopic() +"\", \"operator\": \"and\"}}}");
		}
		if(StringUtils.hasText(condition.getMessage())) {
			bd.append(",{\"match\":{\"message\":{\"query\":\""+ condition.getMessage() +"\", \"operator\": \"or\"}}}");
		}
		bd.append("]}}");
		if(condition.getSize() > 0) {
			bd.append(", \"from\": " + (condition.getPage() - 1) * condition.getSize() +" ,\"size\": " + condition.getSize());
		}
		bd.append(",\"sort\":[");
		bd.append("{\"createTime\": \"desc\"}");
		bd.append("]");
		bd.append("}");
		return bd.toString();
	}
	
	/**
	 * MQTT 建立连接时推送
	 * @param connected 建立连接时消息
	 * @return 处理结果
	 *
	 * @author zhouhui
	 * @since 1.0.0 
	 */
	public R<Object> connectedEvent(MqttConnectedListener connected){
		processMessage(connected);
		
		connected.setMessageType(MqttMessageType.CONNECTED.getType());
		
		//保存MQTT消息信息
		saveMqttMessage(connected);
		
		//更新设备的在线状态
		updateOnlineStatus(connected, 1);
		
		return R.ok();
	}
	
	/**
	 * MQTT 连接断开时推送
	 * @param disconnected 建立断开时消息
	 * @return 处理结果
	 *
	 * @author zhouhui
	 * @since 1.0.0 
	 */
	public R<Object> disconnectedEvent(MqttDisconnectedListener disconnected){
		processMessage(disconnected);
		
		disconnected.setMessageType(MqttMessageType.DISCONNECTED.getType());
		
		//保存MQTT消息信息
		saveMqttMessage(disconnected);
		
		//更新设备的离线状态
		updateOnlineStatus(disconnected, 0);
		
		return R.ok();
	}
	
	/**
	 * MQTT消息订阅
	 * @param subscribed 订阅信息
	 * @return R<Object> 处理结果
	 *
	 * @author zhouhui
	 * @since 1.0.0 
	 */
	public R<Object> subscribedEvent(MqttSubscribedListener subscribed){
		processMessage(subscribed);
		
		subscribed.setMessageType(MqttMessageType.SUBSCRIBED.getType());
		
		//保存MQTT消息信息
		saveMqttMessage(subscribed);
		
		return R.ok();
	}
	
	/**
	 * MQTT取消消息订阅
	 * @param subscribed 取消订阅信息
	 * @return R<Object> 处理结果
	 *
	 * @author zhouhui
	 * @since 1.0.0 
	 */
	public R<Object> unsubscribedEvent(MqttSubscribedListener subscribed){
		processMessage(subscribed);
		
		subscribed.setMessageType(MqttMessageType.UNSUBSCRIBED.getType());
		
		//保存MQTT消息信息
		saveMqttMessage(subscribed);
		
		return R.ok();
	}
	
	/**
	 * MQTT消息投递
	 * @param delivered 投递的消息内容
	 * @return R<Object> 处理结果
	 *
	 * @author zhouhui
	 * @since 1.0.0 
	 */
	public R<Object> deliveredEvent(MqttDeliveredListener delivered){
		if(StringUtils.hasText(delivered.getPayload())) {
			byte[] payloadByte = Base64.getDecoder().decode(delivered.getPayload());
			String str = new String(payloadByte, StandardCharsets.UTF_8);
			str = PATTERN.matcher(str).replaceAll("");
			delivered.setPayload(str);
		}
		
		processMessage(delivered);
		
		delivered.setMessageType(MqttMessageType.DELIVERED.getType());
		
		//保存MQTT消息信息
		saveMqttMessage(delivered);
		
		return R.ok();
	}
	
	/**
	 * MQTT 丢弃消息
	 * @param delivered 丢弃消息内容
	 * @return R<Object> 处理结果
	 *
	 * @author zhouhui
	 * @since 1.0.0 
	 */
	public R<Object> droppedEvent(MqttDroppedListener dropped){
		if(StringUtils.hasText(dropped.getPayload())) {
			byte[] payloadByte = Base64.getDecoder().decode(dropped.getPayload());
			String str = new String(payloadByte, StandardCharsets.UTF_8);
			str = PATTERN.matcher(str).replaceAll("");
			dropped.setPayload(str);
		}
		
		processMessage(dropped);
		
		dropped.setMessageType(MqttMessageType.DROPPED.getType());
		
		//保存MQTT消息信息
		saveMqttMessage(dropped);
		return R.ok();
	}
	
	/**
	 * 保存MQTT服务器推送的事件消息
	 * @param content 消息内容
	 *
	 * @author zhouhui
	 * @since 1.0.0
	 */
	private void saveMqttMessage(MqttBaseListener content) {
		ElasticParam param = new ElasticParam();
		param.setEndpoint(Symbols.FORWARD_SLASH + ElasticsearchIndex.MQTT_LOG_INDEX + "/_doc/" + content.getMessageId());
		param.setMethod(HttpMethod.POST.name());
		
		String body = null;
		try {
			body = MAPPER.writeValueAsString(content);
		} catch (Exception e) {
			e.printStackTrace();
		}
		param.setJsonEntity(body);
		R<Object> result = elasticsearchApi.customOperate(param, FeignInnerEntity.HTTP_INNER_VALUE);
		if(!ResultEnum.SUCCESS.getCode().equals(result.getCode())) {
			throw new CustomException(result.getMessage());
		}
	}
	
	/**
	 * 对接收的消息进行处理
	 * @param baseListener
	 *
	 * @author zhouhui
	 * @since 1.0.0
	 */
	private void processMessage(MqttBaseListener baseListener) {
		if(baseListener.getTimestamp() != null) {
			LocalDateTime createTime = LocalDateTime.ofInstant(Instant.ofEpochMilli(baseListener.getTimestamp()), ZoneId.systemDefault());
			baseListener.setCreateTime(createTime.format(DateTimeFormatter.ISO_DATE_TIME));
		}
		
		if(!StringUtils.hasText(baseListener.getMessageId())) {
			String messageId = UUID.randomUUID().toString();
			baseListener.setMessageId(messageId);
		}
		
		//最后再进行处理
		try {
			baseListener.setMessage(MAPPER.writeValueAsString(baseListener));
		} catch (JsonProcessingException e) {
			e.printStackTrace();
		}
	}
	
	/**
	 * 更新设备的在线状态
	 * @param mqttBase 消息内容 
	 * @param online 在线状态，0离线；1在线
	 *
	 * @author zhouhui
	 * @since 1.0.0
	 */
	private void updateOnlineStatus(MqttBaseListener mqttBase, Integer online) {
		LambdaUpdateWrapper<MqttUser> mqttUserUpdate = new LambdaUpdateWrapper<>();
		mqttUserUpdate.eq(MqttUser::getClientId, mqttBase.getClientId());
		if(StringUtils.hasText(mqttBase.getUsername())) {
			mqttUserUpdate.eq(MqttUser::getUsername, mqttBase.getUsername());
		}
		mqttUserUpdate.set(MqttUser::getOnline, online);
		mqttUserMapper.update(null, mqttUserUpdate);
	}
}
